package cn.minicloud;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
import java.util.Random;

/**
 * https://dev.mysql.com/doc/dev/mysql-server/latest/PAGE_PROTOCOL.html#protocol_overview
 * 1 navicat connect by  127.0.0.1:3306
 * 2 wireshark capture network traffic(loopback)
 * 3 see exchange data weather is ok?
 *

 No.	Time	Source	Destination	Protocol	Length	Info
 113	9.784248	::1	::1	TCP	76	58993 → 3306 [SYN] Seq=0 Win=65535 Len=0 MSS=65475 WS=256 SACK_PERM
 114	9.784282	::1	::1	TCP	76	3306 → 58993 [SYN, ACK] Seq=0 Ack=1 Win=65535 Len=0 MSS=65475 WS=256 SACK_PERM
 115	9.784301	::1	::1	TCP	64	58993 → 3306 [ACK] Seq=1 Ack=1 Win=2618880 Len=0  三次握手建立tcp连接
 118	9.786523	::1	::1	MySQL	147	Server Greeting  proto=10 version=5.7.3-Proxy 服务端响应服务端信息
 119	9.786557	::1	::1	TCP	64	58993 → 3306 [ACK] Seq=1 Ack=84 Win=2618880 Len=0
 120	9.786605	::1	::1	MySQL	195	Login Request user=alice                      客户端发起登录请求
 121	9.786623	::1	::1	TCP	64	3306 → 58993 [ACK] Seq=84 Ack=132 Win=2618624 Len=0
 265	11.132391	::1	::1	MySQL	75	Response  OK
 266	11.132409	::1	::1	TCP	64	58993 → 3306 [ACK] Seq=132 Ack=95 Win=2618880 Len=0
 267	11.132464	::1	::1	TCP	64	58993 → 3306 [FIN, ACK] Seq=132 Ack=95 Win=2618880 Len=0
 268	11.132476	::1	::1	TCP	64	3306 → 58993 [ACK] Seq=95 Ack=133 Win=2618624 Len=0
 274	11.137377	::1	::1	TCP	64	3306 → 58993 [FIN, ACK] Seq=95 Ack=133 Win=2618624 Len=0
 275	11.137412	::1	::1	TCP	64	58993 → 3306 [ACK] Seq=133 Ack=96 Win=2618880 Len=0

 */
public class MySQLHandshakeHandler extends ChannelInboundHandlerAdapter {
    private static final Logger logger = LoggerFactory.getLogger(MySQLHandshakeHandler.class);

    private byte[] scramble = new byte[20];
    private boolean handshakeSent = false;
    private boolean loginReceived = false;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        sendHandshake(ctx);
    }

    private void sendHandshake(ChannelHandlerContext ctx) {
        // MySQL Protocol version 10 handshake
        ByteBuf buf = Unpooled.buffer();
        buf.writeByte(0x0a); // Protocol version
        buf.writeBytes("5.7.3-Proxy\0".getBytes(StandardCharsets.UTF_8)); // Server version
        buf.writeIntLE(1); // Connection ID
        new Random().nextBytes(scramble);
        buf.writeBytes(scramble, 0, 8); // first 8 bytes scramble
        buf.writeByte(0x00);
        buf.writeShortLE(0x0000); // capability flags (lower 2 bytes)
        buf.writeByte(0x21); // charset
        buf.writeShortLE(0x0002); // status flags
        buf.writeShortLE(0xffff); // capability flags (upper 2 bytes)
        buf.writeByte(20); // auth plugin data len
        buf.writeBytes(new byte[10]); // reserved
        buf.writeBytes(scramble, 8, 12); // next part of scramble
        buf.writeByte(0x00);
        buf.writeBytes("mysql_native_password\0".getBytes(StandardCharsets.UTF_8));
        sendPacket(ctx, buf);
        handshakeSent = true;
        logger.info("Sent handshake packet to client");
    }

    private void sendPacket(ChannelHandlerContext ctx, ByteBuf body) {
        int len = body.readableBytes();
        ByteBuf packet = ctx.alloc().buffer(4 + len);
        packet.writeMediumLE(len);
        packet.writeByte(0); // sequence id
        packet.writeBytes(body);
        ctx.writeAndFlush(packet);
    }
    private String username;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (!(msg instanceof ByteBuf)) {
            ctx.fireChannelRead(msg);
            return;
        }
        ByteBuf buf = (ByteBuf) msg;
        if (!loginReceived) {
            loginReceived = true;
            // Extract username from login request according to MySQL protocol
            // Attempt to skip MySQL packet header (3-byte length + 1-byte sequence) if present
            if (buf.readableBytes() > 4 && buf.getByte(buf.readerIndex() + 4) == 0x0a) {
                buf.skipBytes(4); // packet header
            }
            buf.skipBytes(4);  // capability flags
            buf.skipBytes(4);  // max packet size
            buf.skipBytes(1);  // charset
            buf.skipBytes(23); // reserved
            int unameStart = buf.readerIndex();
            int unameLen = 0;
            while (buf.getByte(unameStart + unameLen) != 0) unameLen++;
            byte[] unameBytes = new byte[unameLen];
            buf.getBytes(unameStart, unameBytes);
            username = new String(unameBytes, StandardCharsets.UTF_8);
            buf.readerIndex(unameStart + unameLen + 1);
            logger.info("Handshake payload hex: {}", io.netty.buffer.ByteBufUtil.hexDump(buf));
            logger.info("Accepted login from client, username={}", username);

            sendOk(ctx);
            routeAndProxy(ctx);
        }
        buf.release();
    }

    private void routeAndProxy(ChannelHandlerContext ctx) {
        AuthRouter router = new AuthRouter();
        String backendAddr = router.route(username);
        new BackendSession(backendAddr, ctx.channel()).connectAndStartForwarding();
        ctx.pipeline().remove(this);
    }

    private void sendOk(ChannelHandlerContext ctx) {
        ByteBuf buf = Unpooled.buffer();
        buf.writeByte(0x00); // OK header
        buf.writeByte(0x00); // affected rows
        buf.writeByte(0x00); // last insert id
        buf.writeShortLE(0x0002); // status flags
        buf.writeShortLE(0x0000); // warning count
        sendPacket(ctx, buf);
    }
}
